
#include "RealTimeDealList.h"

#include <core/time/time.h>

#include "../BondMarketStream.h"
#include "../JumpListNew.h"
#include "RealTimeDealCache.h"

#if WIN32
#include <windows.h>
#else
#endif
#include "RealTimeDealStat.h"

#ifndef BASSERTS
#define BASSERTS(smt, fmt, ...)    \
  char buf[256];                   \
  FMTBUF(buf, fmt, ##__VA_ARGS__); \
  Q_ASSERT_X(smt, "", buf);
#endif  // !BASSERTS

#ifndef max
#define max(a, b) (((a) > (b)) ? (a) : (b))
#endif

typedef MarketStreamInfo Record;
namespace detail {
class RecordMap : public IRecordMap {
  typedef std::map<DealIndex, Record> IndexRecordMap;
  IndexRecordMap m_values;

 public:
  virtual ~RecordMap() {}
  virtual void Enumerate(Callback* cb) const {
    IndexRecordMap::const_iterator it = m_values.cbegin();
    for (; it != m_values.cend(); ++it) {
      cb->OnEach(it->first, &it->second);
    }
  }
  virtual const void* FindRecord(const DealIndex& id) const {
    IndexRecordMap::const_iterator it = m_values.find(id);
    if (it == m_values.end()) return nullptr;
    return &it->second;
  }
  virtual void Add(const DealIndex& id, const void* record) {
    const Record* precord = (const Record*)record;
    IndexRecordMap::iterator it = m_values.find(id);
    if (it == m_values.end())
      m_values.insert(IndexRecordMap::value_type(id, *precord));
    else
      it->second = *precord;
  }
  virtual void Remove(const DealIndex& id) {
    // m_values中不能删除,UI中的RealTimeDealModel还保留这对应id
    IndexRecordMap::const_iterator it = m_values.find(id);
    if (it != m_values.end()) {
      m_values.erase(it);
    }
  }
};
class RTDealTimeComparator {
  IRecordMap* m_values;

 public:
  RTDealTimeComparator(IRecordMap* values) : m_values(values) {}
  void SetValues(IRecordMap* values) { m_values = values; }
  RTDealTimeComparator(const RTDealTimeComparator& cmp)
      : m_values(cmp.m_values) {}
  RTDealTimeComparator& operator=(const RTDealTimeComparator& cmp) {
    m_values = cmp.m_values;
    return *this;
  }
  int operator()(const DealIndex& b1, const DealIndex& b2) const {
    // max(obj.m_create_time, obj.m_modify_time) 较大的排在前面
    const void* v1 = m_values->FindRecord(b1);
    const void* v2 = m_values->FindRecord(b2);
    if (v1 && v2) {
      const Record& r1 = *((const Record*)v1);
      const Record& r2 = *((const Record*)v2);
      time_t t1 = max(r1.m_create_time, r1.m_modify_time);
      time_t t2 = max(r2.m_create_time, r2.m_modify_time);
      // 参见MarketStreamInfo的operator<
      return t1 == t2 ? (r1.m_indexID > r2.m_indexID ? -1 : 1)
                      : (t1 > t2 ? -1 : 1);
    }
    if (v1 == nullptr) {
      // BLOGERR("[STREAM][INVALID]can not find DealIndex:%s", b1.id.c_str());
      return 1;
    }
    if (v2 == nullptr) {
      // BLOGERR("[STREAM][INVALID]can not find DealIndex:%s", b2.id.c_str());
      return -1;
    }
    return 0;
  }
  int GetExpandLength(int count, int length) { return length * 2; }
};

typedef JumpList<DealIndex, RTDealTimeComparator> BondDealJumpListBase;
class CBondDealJumpList : public BondDealJumpListBase, public IBatchedKeys {
  struct Operation {
    int indexid : 30;  // MarketStreamInfo.m_indexID
    int optype : 2;    // 操作类型:Null/Add/Erase/Update
    Operation* next;   // 单链表
  };
  typedef std::map<int, Operation*> OperationMap;  // key:batchid;
  int m_batch_id;                                  // 当前的批此id
 public:
  CBondDealJumpList(RTDealTimeComparator* comparator = 0, int stepCount = 40,
                    int hashInitSize = 2560)
      : BondDealJumpListBase(comparator, stepCount, hashInitSize),
        m_batch_id(-1) {}
  virtual int GetBatchId() const { return m_batch_id; }
  virtual void SetBatchId(int id) {
    if (id >= m_batch_id) m_batch_id = id;
  }
  virtual void GetKeys(std::vector<DealIndex>& allkeys) const {
    allkeys.reserve(GetCount());
    return GetAllKeys(allkeys);
  }
};

class BondKeyDealIndexMap {
  typedef std::string BondKeyType;
  class DealIndexSet : protected std::set<DealIndex> {
    DealIndex m_head;  // >1的时候有效
    bool m_dirty;      // GetHead获取的时候需要更新
   public:
    DealIndexSet() : m_dirty(true) {}
    void Add(const DealIndex& id, IRecordMap* values) {
      insert(id);
      m_dirty = true;
    }
    void Remove(const DealIndex& id, IRecordMap* values) {
      iterator it = find(id);
      if (it != end()) erase(it);
      m_dirty = true;
    }
    bool GetHead(DealIndex& id, IRecordMap* values) {
      if (size() > 1 && m_dirty) {
        UpdateHead(values);
      }
      m_dirty = false;
      if (empty()) return false;
      if (size() == 1) {
        id = *begin();
        return true;
      }
      id = m_head;
      BASSERTS(find(id) != end(), "");
      return true;
    }

   protected:
    void UpdateHead(IRecordMap* values) {
      if (size() <= 1 || values == nullptr) return;
      int index = 0;
      const void* last = nullptr;
      for (iterator it = begin(); it != end(); ++it, ++index) {
        const DealIndex& cur = *it;
        const void* cur_record = values->FindRecord(cur);
        if (cur_record) {
          if (!last)  // 第一个
          {
            m_head = cur;
            last = cur_record;
          } else  // 挑选排列在最前面的,即时间最大,m_indexID最大
          {
            const Record& r1 = *((const Record*)last);
            const Record& r2 = *((const Record*)cur_record);
            time_t t1 = max(r1.m_create_time, r1.m_modify_time);
            time_t t2 = max(r2.m_create_time, r2.m_modify_time);
            if (t2 > t1 || (t2 == t1 && r2.m_indexID > r1.m_indexID)) {
              m_head = cur;
              last = cur_record;
            }
          }
        }
      }
      BASSERTS(find(m_head) != end(), "");
    }
  };
  typedef std::map<BondKeyType, DealIndexSet*> BondIndexMap;
  BondIndexMap m_index_map;

 public:
  BondKeyDealIndexMap() {}
  ~BondKeyDealIndexMap() {
    for (BondIndexMap::iterator it = m_index_map.begin();
         it != m_index_map.end(); ++it) {
      DealIndexSet* diset = it->second;
      if (diset) delete diset;
    }
    m_index_map.clear();
  }
  void Add(const char* szBondKey, const char* szListedMarket,
           const char* szCompanyId, const DealIndex& id, IRecordMap* values) {
    char keybuf[128] = {0};
    FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
    BondIndexMap::iterator it = m_index_map.find(keybuf);
    DealIndexSet* diset = nullptr;
    if (it == m_index_map.end()) {
      diset = new DealIndexSet();
      m_index_map.insert(BondIndexMap::value_type(keybuf, diset));
    } else
      diset = it->second;
    diset->Add(id, values);
  }
  void Remove(const char* szBondKey, const char* szListedMarket,
              const char* szCompanyId, const DealIndex& id,
              IRecordMap* values) {
    char keybuf[128] = {0};
    FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
    BondIndexMap::iterator it = m_index_map.find(keybuf);
    if (it != m_index_map.end()) {
      DealIndexSet* diset = it->second;
      diset->Remove(id, values);
    }
  }
  bool Find(const char* szBondKey, const char* szListedMarket,
            const char* szCompanyId, DealIndex& id, IRecordMap* values) const {
    char keybuf[128] = {0};
    FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
    BondIndexMap::const_iterator it = m_index_map.find(keybuf);
    if (it == m_index_map.end()) return false;

    // 找出diset的几个之中,time最大的那个
    const DealIndexSet* diset = it->second;
    bool suc = const_cast<DealIndexSet*>(diset)->GetHead(id, values);
    return suc;
  }
};
RealTimeDealList::RealTimeDealList(DayType day, int64 state)
    : m_tag(OBJECT_TAG),
      m_day(day),
      m_bonds(nullptr),
      m_check(false),
      m_state(state) {
  m_values = new detail::RecordMap();
  m_keys = new detail::CBondDealJumpList();
  m_stats = new detail::RealTimeDealStat();
  m_cache = new detail::RealTimeDealCache(m_state);
  m_bonds = new BondKeyDealIndexMap();

  // 必须设置比较器,否则m_keys无法插入数据,插入会崩溃
  m_keys->SetComparator(new detail::RTDealTimeComparator(m_values));
}
RealTimeDealList::~RealTimeDealList() {
  if (m_values) delete m_values;
  if (m_keys) delete m_keys;
  if (m_stats) delete m_stats;
  if (m_cache) delete m_cache;
  if (m_bonds) delete m_bonds;
  m_values = nullptr;
  m_keys = nullptr;
  m_stats = nullptr;
  m_cache = nullptr;
  m_bonds = nullptr;
}

int RealTimeDealList::GetRecords(IRealTimeDealModel* keys, int& pos, int count,
                                 void* records) {
  // UI线程:此接口用于界面表格的一屏幕记录
  // 当日成交中还有一个需要访问所有记录,判断是否是只有一个broker;
  // 这种需求需要访问所有记录,重复计算的是哪一部分?
  // 不管怎么样,始终获取count个记录,
  // 如果范围段[pos,pos+count)中有DealIndex不存在,那么往前往后继续遍历

  // 1.首先,将keys与m_keys同步,如果严重滞后,则重建并过滤
  bool synchronized = false;
  int nOldCount = keys->GetCount();
  class SyncCallback : public detail::RealTimeDealCache::Callback {
    IRealTimeDealModel* m_keys;

   public:
    SyncCallback(IRealTimeDealModel* keys) : m_keys(keys) {}
    virtual bool OnRecord(int id, const Record* record) {
      m_keys->AddRecord(id, record);
      return true;
    }
  };
  class FlashCallback : public detail::RealTimeDealCache::Callback {
    IRealTimeDealModel* m_keys;

   public:
    FlashCallback(IRealTimeDealModel* keys) : m_keys(keys) {}
    virtual bool OnRecord(int id, const Record* record) {
      m_keys->AddFlash(id, record);
      return true;
    }
  };

  int batchid = keys->GetBatchId();
  if (!IsObjectValid()) return 0;

  bool equal_state = keys->GetState() == m_cache->GetState();
  if (equal_state && batchid >= 0 && batchid == m_cache->GetEndId()) {
    // 已经是同步的,无需再次同步
  } else if (equal_state && batchid >= 0 && batchid >= m_cache->GetStartId() &&
             batchid < m_cache->GetEndId()) {
    // 快速同步已经做改进,将没有更新的合并,并分为两类:删除和添加,并维持原有顺序,
    // 然后先删除所有该删除的,再添加所有该添加的
    SyncCallback callback(keys);
    keys->StartCompare(m_values);
    m_cache->GetSince(batchid, callback);
  } else  // 全量同步
  {
    keys->SetState(m_cache->GetState());
    keys->ResetKeys();
    synchronized = true;
    keys->SyncKeys(m_keys, m_values);  // 比较耗时,
    if (equal_state && batchid >= 0 &&
        batchid <=
            m_cache
                ->GetEndId())  // batchid为-1表示全部重拉,无需重建闪烁,全部清空闪烁
    {
      FlashCallback flashcb(keys);
      m_cache->GetSince(batchid, flashcb);
    } else
      keys->ClearFlash();
  }

  // 2.更新获取行位置,快速
  int nNewCount = keys->GetCount();
  if (nNewCount != nOldCount || synchronized) {
    // pos范围需要重新调整,直至满足要求pos+count<nNewCount
    // nNewCount 是最新数据的总个数,count是界面表格能显示的个数
    // 返回值是实际获取到的记录个数,肯定<=count
  }
  int nret = 0;
  DealIndex indexes[512];  // 一屏幕最大显示多少行,512是绝对足够的

  DealIndex* pIndex = nullptr;
  bool bNeedDel = false;
  if (count < 512) {
    // 3.获取一屏幕的记录,通常只有30-50个左右,很快
    pIndex = indexes;
  } else {
    bNeedDel = true;

    pIndex = new DealIndex[count];
  }

  int nadded = 0;  // 通常最终返回,nadded==nret
  keys->StartCompare(m_values);
  nret = keys->GetKeys(pos, count, indexes);

  Record* precords = (Record*)records;
  for (int i = 0; i < nret; i++) {
    const Record* record = (const Record*)m_values->FindRecord(indexes[i]);
    BASSERTS(record != nullptr, "m_values can not find key:%d",
             indexes[i].GetHash());
    if (record) precords[nadded++] = *record;
  }

  // 4.检查是否有全部记录的特殊标志:多个broker,用于控制broker列的显示和隐藏
  //  比较耗时,需要访问所有的values(过滤后的),但是只有在OnFilter中,才会将multidirty设置为true
  //  synchronized 为false时,已经通过GetSince,AddRecord来更新到m_multitest了;
  if (synchronized && keys->IsMultiEnable()) {
    keys->CheckMulti(m_values);
  }
  if (bNeedDel) delete[] pIndex;
  return nadded;
}

IRecordMap* RealTimeDealList::GetValues() { return m_values; }
IRecordMap* RealTimeDealList::GetKeys(std::vector<DealIndex>& keys,
                                      bool all) const {
  // 获取当前的所有成交id的数组
  // all:false 表示获取排除作废成交的成交id的数组
  keys.clear();
  if (all) {
    m_keys->GetKeys(keys);
  } else {
    std::vector<DealIndex> allkeys;
    m_keys->GetKeys(allkeys);
    keys.reserve(allkeys.size());
    for (int i = 0; i < (int)allkeys.size(); i++) {
      const MarketStreamInfo* msi =
          (const MarketStreamInfo*)m_values->FindRecord(allkeys[i]);
      if (msi && !msi->IsDealStatusCanceled()) keys.push_back(allkeys[i]);
    }
  }
  return m_values;
}
int RealTimeDealList::GetVersion(int companyID) const {
  std::map<int, int>::const_iterator it = m_mapversion.find(companyID);
  if (it == m_mapversion.end()) return 0;
  return it->second;
}
void RealTimeDealList::InsertDirectly(const MarketStreamInfo& info,
                                      uint32_t unDay, int batchid) {
  BASSERTS(unDay == m_day, "[DEAL]m_day=%d  unDay:%d", m_day, unDay);
  int nDealStatus = atoi(info.m_body.m_dealStatus);
  /*BLOGINFO("[STREAM][RECV]m_indexID:%d	m_id:%s  count:%d code:%s key:%s
     market:%s status:%s mtime:%d", info.m_indexID, info.m_body.m_id,
     m_keys?m_keys->GetCount():0,
          info.m_body.m_bondCode,info.m_body.m_bondkey,info.m_body.m_listedmarket,info.m_body.m_dealStatus,info.m_modify_time
          );
          */
  DealIndex id(info.m_body.m_id);
  /*
          此处更新涉及如下5个模块:
          m_keys	 :快速有序队列
          m_values :成交记录表
          m_stats  :成交统计表
          m_bonds  :债券到成交索引的表
          m_cache  :推送命令缓存

          m_keys(快速有序队列):
          是一个存储MarketStreamType2Body.m_id的有序队列,顺序按照RTDealTimeComparator来比较
          具体的就是MarketStreamInfo的时间m_modify_time|m_create_time较大的排在前面,
          时间相同,再比较m_indexID,较大的排在前面
          快速查找:哈希定位,复杂度O(1)
          删除移除:log(N)
          插入操作:Insert:先删除,后找到合适的排序位置,插入,要求键值对应的记录必须到位
          依赖m_values事先更新对应键的值,否则插入位置依旧是原来的位置

          m_values(成交记录表):
          底层实现是map<IndexID,MarketStreamInfo>,记录所有成交的最新记录信息

          m_stats(成交统计表):

          m_bonds(债券到成交索引的表):
          每个债券可能有多个成交,这多个成交记录的DealIndex构成一个集合
          m_bonds保存每个债券的所有成交,并维护每个债券的最新成交的DealIndex

          3.5.0中新增作废成交,处理逻辑如下:

          */
  bool stat_erase = nDealStatus == 8 || nDealStatus >= 4;
  if (m_keys->IsExist(id)) {
    const void* ptr = m_values->FindRecord(id);
    BASSERTS(ptr, "");
    if (ptr) {
      const Record& rd = *((const Record*)ptr);
      if (stat_erase)
        m_stats->EraseStat(rd, true);
      else
        m_stats->UpdateStat(rd, info);
    } else if (!stat_erase) {
      // BLOGERR("[STREAM]invalid DealIndex:%s,does not exist in m_values",
      // id.id.c_str());
      m_stats->AddStat(info);
    }
  } else {
    if (stat_erase)
      m_stats->EraseStat(info, false);
    else
      m_stats->AddStat(info);
  }

  if (nDealStatus == 8 || nDealStatus < 4) {
    m_values->Add(id, &info);
    m_keys->Insert(id);
  } else {
    m_keys->Erase(id);
    m_values->Remove(id);
  }

  // 如果该成交是作废成交，那么也应该从对应债券的成交id队列中删除，即Remove，而不是之前的Add
  if (nDealStatus < 4)
    m_bonds->Add(info.m_body.m_bondkey, info.m_body.m_listedmarket,
                 info.m_company_id, id, m_values);
  else
    m_bonds->Remove(info.m_body.m_bondkey, info.m_body.m_listedmarket,
                    info.m_company_id, id, m_values);
  m_keys->SetBatchId(batchid);
  m_cache->Add(batchid, &info);
  int companyID = atoi(info.m_company_id);
  std::map<int, int>::iterator it = m_mapversion.find(companyID);
  if (it == m_mapversion.end()) {
    m_mapversion.insert(std::make_pair(companyID, info.m_indexID));
  } else {
    if (info.m_indexID > it->second) it->second = info.m_indexID;
  }
  return;
}

bool RealTimeDealList::GetStat(const std::string& bondKey,
                               MarketStreamRec& stat) {
  return m_stats->GetStat(bondKey, stat);
}

bool RealTimeDealList::IsObjectValid() const {
  if (m_cache == nullptr || m_keys == nullptr || m_stats == nullptr ||
      m_values == nullptr)
    return false;
  return m_tag == OBJECT_TAG;
}
bool RealTimeDealList::ScanForRecord(const char* szBondKey,
                                     const char* szListedMarket,
                                     const char* szCompanyId, time_t t,
                                     MarketStreamInfo& record) const {
  int nMin = 0x7FFFFFFF;
  bool corp_empty = szCompanyId == nullptr || *szCompanyId == '\0';
  // vector<DealIndex> indexes;
  // GetKeys(indexes);
  detail::CBondDealJumpList::KeyIterator it = m_keys->GetBegin();
  // for (int k = 0; k < (int)indexes.size(); k++)
  for (; it != m_keys->GetEnd(); ++it) {
    const void* ptr = m_values->FindRecord(*it /*indexes[k]*/);
    if (ptr == nullptr) continue;
    const Record& rd = *((const Record*)ptr);
    if (strcmp(rd.m_body.m_bondkey, szBondKey) == 0 &&
        strcmp(rd.m_body.m_listedmarket, szListedMarket) == 0 &&
        (corp_empty || strcmp(rd.m_company_id, szCompanyId) == 0)) {
      if (t <= 0) {
        record = rd;
        return true;
      } else if (rd.m_body.m_createTime <= t)  // 大于当前的时间
      {
        int nDelta = t - rd.m_body.m_createTime;
        if (nDelta < nMin) {
          nMin = nDelta;
          record = rd;
          return true;
        }
      }
    }
  }
  return false;
}
bool RealTimeDealList::FindRecord(const char* szBondKey,
                                  const char* szListMarket,
                                  const char* szCompanyId,
                                  MarketStreamInfo& record, time_t t) const {
  // 一个债券可能有多笔成交,rdlist是时间倒序排列,根据时间t来选取一个成交数据
  // ScanForRecord 是
  bool corp_empty = szCompanyId == nullptr || *szCompanyId == '\0';
  if (t <= 0 && !corp_empty && m_bonds) {
    DealIndex dindex;
    const void* ptr = nullptr;
    bool suc = false;
    if (m_bonds->Find(szBondKey, szListMarket, szCompanyId, dindex, m_values)) {
      ptr = m_values->FindRecord(dindex);
      if (ptr) {
        record = *((const Record*)ptr);
        suc = true;
      }
    }
    if (m_check) {
      MarketStreamInfo msi;
      bool suc_scan =
          ScanForRecord(szBondKey, szListMarket, szCompanyId, t, msi);
      BASSERTS(suc == suc_scan, "");
      if (suc_scan && suc) {
        BASSERTS(memcmp(&msi, &record, sizeof(MarketStreamInfo)) == 0, "");
      }
    }
    return suc;
  }
  return ScanForRecord(szBondKey, szListMarket, szCompanyId, t, record);
}

MarketStreamInfo* RealTimeDealList::ElementAt(DealIndex key) {
  MarketStreamInfo* pInfo = (MarketStreamInfo*)(m_values->FindRecord(key));
  return pInfo;
}
//////////////////////////////////////////////////////////////////////////////////
// class MultiDayDealList : public RealTimeDealList

MultiDayDealList::MultiDayDealList(uint32_t dayBegin, uint32_t dayEnd)
    : RealTimeDealList(dayBegin, 0) {
  m_dayBegin = dayBegin;
  m_dayEnd = dayEnd;
  if (m_dayBegin > 0 && m_dayEnd > 0 && m_dayBegin <= m_dayEnd) {
  }
}

MultiDayDealList::~MultiDayDealList() {}

uint32_t MultiDayDealList::GetDayBegin() { return m_dayBegin; }
uint32_t MultiDayDealList::GetDayEnd() { return m_dayEnd; }

bool MultiDayDealList::CheckDayRange(uint32_t dayBegin, uint32_t dayEnd) {
  if (m_dayBegin == dayBegin && m_dayEnd == dayEnd) return true;
  return false;
}

void MultiDayDealList::ChangeDayRange(uint32_t beginDay, uint32_t endDay) {
  m_dayBegin = beginDay;
  m_dayEnd = endDay;
}

void MultiDayDealList::InsertDirectly(const MarketStreamInfo& info,
                                      uint32_t unDay, int batchid) {
  if (unDay > m_dayEnd || unDay < m_dayBegin) return;

  int nDealStatus = atoi(info.m_body.m_dealStatus);
  DealIndex id(info.m_body.m_id);

  bool stat_erase = nDealStatus == 8 || nDealStatus >= 4;
  if (m_keys->IsExist(id)) {
    const void* ptr = m_values->FindRecord(id);
    BASSERTS(ptr, "");
    if (ptr) {
      const Record& rd = *((const Record*)ptr);
      if (stat_erase)
        m_stats->EraseStat(rd, true);
      else
        m_stats->UpdateStat(rd, info);
    } else if (!stat_erase) {
      // BLOGERR("[STREAM]invalid DealIndex:%s,does not exist in m_values",
      // id.id.c_str());
      m_stats->AddStat(info);
    }
  } else {
    if (stat_erase)
      m_stats->EraseStat(info, false);
    else
      m_stats->AddStat(info);
  }

  if (nDealStatus == 8 || nDealStatus < 4) {
    m_values->Add(id, &info);
    m_keys->Insert(id);
  } else {
    m_keys->Erase(id);
    m_values->Remove(id);
  }

  // 如果该成交是作废成交，那么也应该从对应债券的成交id队列中删除，即Remove，而不是之前的Add
  if (nDealStatus < 4)
    m_bonds->Add(info.m_body.m_bondkey, info.m_body.m_listedmarket,
                 info.m_company_id, id, m_values);
  else
    m_bonds->Remove(info.m_body.m_bondkey, info.m_body.m_listedmarket,
                    info.m_company_id, id, m_values);
  m_keys->SetBatchId(batchid);
  m_cache->Add(batchid, &info);
  return;
}
}  // namespace detail
